package consumer

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/zeromicro/go-zero/core/logx"
	"im_sys/apps/im/immodels"
	"im_sys/apps/im/ws/ws_server"
	"im_sys/apps/task/mq/internal/svc"
	"im_sys/apps/task/mq/mq_struct"
	"im_sys/pkg/constants"
)

type ChatMsgConsumer struct {
	svc *svc.ServiceContext
	logx.Logger
}

func NewChatMsgConsumer(svc *svc.ServiceContext) *ChatMsgConsumer {
	return &ChatMsgConsumer{
		svc:    svc,
		Logger: logx.WithContext(context.Background()),
	}
}

// Consume 作为kafka消费者获取消息，将消息存储到MongoDB数据库，通过WebSocket客户端将消息转发给接收者
func (c *ChatMsgConsumer) Consume(ctx context.Context, key, value string) error {
	fmt.Println(key, value)

	var data mq_struct.ChatMsg

	// 从kafka中得到发送者的聊天消息，进行反序列化
	if err := json.Unmarshal([]byte(value), &data); err != nil {
		return err
	}

	// 将聊天消息写入数据库
	if err := c.svc.ChatLogModel.Insert(ctx, &immodels.ChatLog{
		ConversationId: data.ConversationId,
		SendId:         data.SendId,
		RecvId:         data.RecvId,
		MsgFrom:        0,
		ChatType:       data.ChatType,
		MsgType:        data.MType,
		MsgContent:     data.Content,
		SendTime:       data.SendTime,
	}); err != nil {
		return err
	}

	// 通过websocket的客户端，将消息发给websocket服务端，服务端将消息转发给接收者
	return c.svc.WsClient.Send(ws_server.Message{
		FrameType: ws_server.FrameData,
		Method:    "msg_push",
		FromId:    constants.SystemRootUid,
		Data:      data,
	})
}
